package com.ionehe.group.common.async.service;

import com.google.common.base.Throwables;
import com.ionehe.group.common.async.model.Message;
import com.ionehe.group.common.async.mq.AsyncMQProducer;
import com.ionehe.group.common.async.util.AsyncSerializeUtil;
import com.ionehe.group.common.async.util.Json;
import com.ionehe.group.common.notify.message.exception.AsyncException;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;


/**
 * Copyright (c) 2021 ionehe.com
 * Date: 2021/4/21
 * Time: 1:22 下午
 *
 * @author 2021年 <a href="mailto:a@ionehe.com">秀</a>
 * 【异步消息核心处理器】
 */
@Service
@Data
@Slf4j
public class AsyncHandle implements AsyncServer, AsyncClient {

    @Autowired
    private MessageHandle messageHandle;

    @Autowired
    private AsyncMQProducer asyncMQProducer;

    @Value("${notify.vendor:rocketmq}")
    private String vendor;

    private static Long SEC = Long.valueOf(1000L);
    private static Long MIN = Long.valueOf(60000L);
    private static Long HOUR = Long.valueOf(3600000L);


    private Long[] delayLevels = new Long[]{Long.valueOf(1L * SEC.longValue()), Long.valueOf(5L * SEC.longValue()), Long.valueOf(10L * SEC.longValue()), Long.valueOf(30L * SEC.longValue()),
            Long.valueOf(1L * MIN.longValue()), Long.valueOf(2L * MIN.longValue()), Long.valueOf(3L * MIN.longValue()), Long.valueOf(4L * MIN.longValue()), Long.valueOf(5L * MIN.longValue()), Long.valueOf(6L * MIN.longValue()), Long.valueOf(7L * MIN.longValue()), Long.valueOf(8L * MIN.longValue()), Long.valueOf(9L * MIN.longValue()), Long.valueOf(10L * MIN.longValue()), Long.valueOf(20L * MIN.longValue()), Long.valueOf(30L * MIN.longValue()),
            Long.valueOf(1L * HOUR.longValue()), Long.valueOf(2L * HOUR.longValue())};


    @Override
    public void register(Object object) {
        try {
            this.messageHandle.putMethod(object);
        } catch (Exception e) {
            log.error("async register fail cause:{}", Throwables.getStackTraceAsString(e));
        }
    }


    @Override
    public void sendDelay(String tag, Long delayTime, Object... objects) {
        if ("rocketmq".equals(this.vendor)) {
            int delayLevel = transfer(delayTime);
            sendRocketDelay("other", tag, delayLevel, objects);
        }
    }


    @Override
    public void sendQuick(String tag, Object... objects) {
        send("quick", tag, objects);
    }


    @Override
    public void sendSlow(String tag, Object... objects) {
        send("slow", tag, objects);
    }

    private int transfer(Long delayTime) {
        int level = 0;
        for (int i = 0; i < this.delayLevels.length; i++) {
            Long delayLevel = this.delayLevels[i];
            if (delayTime <= delayLevel) {
                level = i;
                break;
            }
        }
        return level + 1;
    }

    private void send(String key, String tag, Object... object) {
        try {
            Message message = new Message();
            message.setTag(tag);
            message.setBody(AsyncSerializeUtil.serialize(object));
            this.asyncMQProducer.send(key, Json.of(message), "async");
            log.info("async mq send message success,tag:{},objects:{}", tag, object);
        } catch (Exception e) {
            log.error("async mq send fail cause:{},body:{}", Throwables.getStackTraceAsString(e), object);
            throw new AsyncException("async mq send fail !!!");
        }
    }


    private void sendRocketDelay(String key, String tag, int delayLevel, Object... object) {
        try {
            Message message = new Message();
            message.setTag(tag);
            message.setBody(AsyncSerializeUtil.serialize(object));
            this.asyncMQProducer.send(key, Json.of(message), "async", delayLevel);
            log.info("async mq send message success,tag:{},objects:{},delayTime:{}",tag, object, delayLevel);
        } catch (Exception e) {
            log.error("async mq send fail cause:{},body:{}", Throwables.getStackTraceAsString(e), object);
            throw new AsyncException("async mq send fail !!!");
        }
    }
}


